package com.atguigu.gmall.realtime.function;


import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lzc
 * @Date 2023/1/10 13:56
 */
public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T, T> {
    
    public abstract String getTable();
    public abstract String getId(T input);
    
    public abstract void addDim(T input, JSONObject dim);
    
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取线程池
        threadPool = ThreadPoolUtil.getTreadPool();
        
    }
    
    @Override
    public void asyncInvoke(T input,   // 流中的每个元素
                            ResultFuture<T> resultFuture) {
        // 每来一条数据, 执行一个子线程
        // 在子线程内完成 dim 的补充
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 每个线程分别获取一个 redis 的客户端和 phoenix 连接
                try {
                    Jedis redisClient = RedisUtil.getRedisClient();
                    DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
                    JSONObject dim = DimUtil.readDim(redisClient, phoenixConn, getTable(), getId(input));
                    // 得到维度, 从 dim 中取值, 给 input 中的相应的字段赋值
                    addDim(input, dim);
                    // input 中的维度信息已经补充完毕,把结果输出
                    resultFuture.complete(Collections.singletonList(input));
                    
                    // 关闭客户端(连接池, 归还)
                    redisClient.close();
                    JdbcUtil.closeConnection(phoenixConn);
                    
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
}
